package com.erpoa.service;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;

/**
 * @author japhet_jiu
 * @version 1.0
 */
@Service
public class MsgService implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
    private static final Logger logger = LoggerFactory.getLogger(MsgService.class);


    @Autowired
    private RabbitTemplate rabbitTemplate;


    @PostConstruct
    public void init() {
        /**
         * 确保消息发送失败后可以重新返回到队列中
         * 注意：yml需要配置 publisher-returns: true
         */
        rabbitTemplate.setMandatory(true);

        /**
         * 消费者确认收到消息后，手动ack回执回调处理
         */
        rabbitTemplate.setConfirmCallback(this);

        /**
         * 消息投递到队列失败回调处理
         */
        rabbitTemplate.setReturnCallback(this);


        // 针对网络原因导致连接断开，利用retryTemplate重连10次
        RetryTemplate retryTemplate = new RetryTemplate();
        retryTemplate.setRetryPolicy(new SimpleRetryPolicy(10));
        rabbitTemplate.setRetryTemplate(retryTemplate);


    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack) {
            //   System.out.println("确认了这条消息："+correlationData);
        } else {
            System.out.println("确认失败了：" + correlationData + "；出现异常：" + cause);
        }
    }


    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        logger.info("returnedMessage ===> replyCode={} ,replyText={} ,exchange={} ,routingKey={}", replyCode, replyText, exchange, routingKey);
        System.out.println("这条消息发送失败了" + message.toString() + "returnedMessage ===> replyCode={} ,replyText={} ,exchange={} ,routingKey={}" + replyCode + replyText + exchange + routingKey);
    }

    /**
     * 发送普通消息到消费队列
     *
     * @param queue
     * @param msg
     */
    public void sendMsg(String queue, String msg) {
        rabbitTemplate.convertAndSend(queue, msg);
    }

    /**
     * 发送延时消息到mq
     *
     * @param exchange  死信交换机
     * @param routeKey  路由key
     * @param data      发送数据
     * @param delayTime 过期时间，单位毫秒
     */
    public void sendDelayMsgToMQ(String exchange, String routeKey, String data, int delayTime) {
        rabbitTemplate.convertAndSend(exchange, routeKey, data, message -> {
            message.getMessageProperties().setExpiration(delayTime + "");
            return message;
        });
    }
}
